package leaf.cloud.common.core.utils.async;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.concurrent.BasicThreadFactory;
import org.springframework.transaction.support.TransactionSynchronizationAdapter;
import org.springframework.transaction.support.TransactionSynchronizationManager;

import javax.validation.ConstraintViolationException;
import java.util.concurrent.*;

/**
 * @ClassName AsyncProcessQueue
 * @Description
 * @Author YeZiHao
 * @Date 2022/7/13 18:25
 **/
@Slf4j
public class AsyncProcessQueue {
    private AsyncProcessQueue() {
    }

    public static AsyncProcessQueue getInstance() {
        return instance;
    }

    @SuppressWarnings("AlibabaConstantFieldShouldBeUpperCase")
    private static final AsyncProcessQueue instance = new AsyncProcessQueue();

    /**
     * 默认最大并发数:cpu 核心 x 2<br>
     */
    private static final int DEFAULT_MAX_CONCURRENT = Runtime.getRuntime().availableProcessors() * 2;

    /**
     * 线程池名称格式
     */
    private static final String THREAD_POOL_NAME = "AsyncProcessPool-%d";

    /**
     * 线程工厂名称
     */
    private static final ThreadFactory FACTORY = new BasicThreadFactory.Builder().namingPattern(THREAD_POOL_NAME)
            .daemon(true).build();

    /**
     * 默认队列大小
     */
    private static final int DEFAULT_SIZE = 1024;

    /**
     * 默认线程存活时间
     */
    private static final long DEFAULT_KEEP_ALIVE = 60L;

    /**
     * NewEntryServiceImpl.java:689
     * Executor
     */
    private static ExecutorService executor;

    /**
     * 执行队列
     */
    private static BlockingQueue<Runnable> executeQueue = new ArrayBlockingQueue<>(DEFAULT_SIZE);

    public void init() {
        // 创建 Executor
        // 此处默认最大值改为处理器数量的 4 倍
        try {
            executor = new ThreadPoolExecutor(DEFAULT_MAX_CONCURRENT, DEFAULT_MAX_CONCURRENT * 4, DEFAULT_KEEP_ALIVE,
                    TimeUnit.SECONDS, executeQueue, FACTORY);
            // 关闭事件的挂钩
            Runtime.getRuntime().addShutdownHook(new Thread(() -> {
                log.info("AsyncProcessor shutting down.");
                executor.shutdown();
                try {
                    // 等待1秒执行关闭
                    if (!executor.awaitTermination(1, TimeUnit.SECONDS)) {
                        log.error("AsyncProcessor shutdown immediately due to wait timeout.");
                        executor.shutdownNow();
                    }
                } catch (InterruptedException e) {
                    log.error("AsyncProcessor shutdown interrupted.");
                    executor.shutdownNow();
                }
                log.info("AsyncProcessor shutdown complete.");
            }
            ));
        } catch (Exception e) {
            log.error("AsyncProcessor init error.", e);
            throw new ExceptionInInitializerError(e);
        }
        log.info("AsyncProcessor has init...");
    }

    /**
     * Task 包装类<br>
     * 此类型的意义是记录可能会被 Executor 吃掉的异常<br>
     */
    public static class TaskWrapper implements Runnable {
        private final Runnable gift;

        public TaskWrapper(final Runnable target) {
            this.gift = target;
        }

        @Override
        public void run() {
            // 捕获异常，避免在 Executor 里面被吞掉了
            if (gift != null) {
                try {
                    gift.run();
                } catch (Exception e) {
                    log.error("TaskWrapper target execute exception.", e);
                }
            }
        }
    }

    /**
     * 执行任务，不管是否成功<br>
     * 其实也就是包装以后的execute方法
     *
     * @param task
     * @return
     */
    public boolean executeTask(Runnable task) {

        try {
            executor.execute(new TaskWrapper(task));
        } catch (RejectedExecutionException e) {
            log.error("Task executing was rejected.", e);
            return false;
        }

        return true;
    }

    /**
     * 提交任务，并可以在稍后获取其执行情况<br>
     * 当提交失败时，会抛出不支持处理的异常
     *
     * @param task
     * @return
     */
    public <T> Future<T> submitTask(Callable<T> task) {
        try {
            return executor.submit(task);
        } catch (RejectedExecutionException e) {
            log.error("Task executing was rejected.", e);
            throw new UnsupportedOperationException("Unable to submit the task, rejected.", e);
        }
    }

    /**
     * 事务提交后异步操作
     * 可指定线程池
     *        //ThreadPoolTaskExecutor executor = ApplicationContextProvider.getBean(threadPoolName,ThreadPoolTaskExecutor.class);
     * @param
     * @param supplier 执行任务
     */
    public void asyncAfterCommit(VoidSupplier supplier) {
        if (!TransactionSynchronizationManager.isSynchronizationActive()) {
            executor.execute(supplier::get);
            return;
        }
        TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {
            @Override
            public void afterCommit() {
                executor.execute(() -> {
                    try {
                        supplier.get();
                    } catch (ConstraintViolationException e) {
                        log.error("param validation error: {}", e);
                    } catch (Throwable e) {
                        log.error("Unexpected error occurred invoking thread pool execute", e);
                    }
                });
            }
        });
    }
}
